package com.ctbri.manage.bydeequ.calculate

import com.ctbri.manage.quality.config.YamlReader
//import com.ctbri.manage.bydeequ.calculate.CalculateV1
//import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.yaml.snakeyaml.Yaml

import java.io.{File, FileInputStream}
import java.util
import java.util.Map
/**
 * @author songyunlong
 * @createTime 2023/6/7 09:31
 * @description
 */
case class DeequCalculate(basicConfigPath: String, dataSource: String){
    YamlReader.getInstance(this.basicConfigPath).setConfForScala(new Yaml().load(new FileInputStream(new File(this.basicConfigPath))).asInstanceOf[util.Map[String, AnyRef]])
    val conf: util.Map[String, AnyRef] = YamlReader.getInstance(basicConfigPath).getConfForScala
    println("file path of basic config:")
    println(this.conf)
    conf.forEach((key, value) => println(key + ": " + conf.get(key)))
    //获取需要执行的功能名和功能所需要的参数名
    val appName = YamlReader.getInstance(this.basicConfigPath).getString("appName")
    val masterSet: String = YamlReader.getInstance(this.basicConfigPath).getString("masterSet")
    val logLevel: String = YamlReader.getInstance(this.basicConfigPath).getString("logLevel")
    val dataSavePath: String = YamlReader.getInstance(this.basicConfigPath).getString("dataSavePath")
    val dataSaveStyle: String = YamlReader.getInstance(this.basicConfigPath).getString("dataSaveStyle")
    val dataExportPath: String = YamlReader.getInstance(this.basicConfigPath).getString("dataExportPath")
    val url: String = YamlReader.getInstance(this.basicConfigPath).getString("url")
    val driver: String = YamlReader.getInstance(this.basicConfigPath).getString("driver")
    val user: String = YamlReader.getInstance(this.basicConfigPath).getString("user")
    val password: String = YamlReader.getInstance(this.basicConfigPath).getString("password")
    val dbtable: String = YamlReader.getInstance(this.basicConfigPath).getString("dbtable")
    val databaseName: String = YamlReader.getInstance(this.basicConfigPath).getString("databaseName")
    val tableName: String = YamlReader.getInstance(this.basicConfigPath).getString("tableName")
    val configPath: String = YamlReader.getInstance(this.basicConfigPath).getString("configPath")
    val needTransformDataType: Boolean = YamlReader.getInstance(this.basicConfigPath).getBoolean("needTransformDataType", false)
    var dataTypeTransformConfigPath: String = null
    if (this.needTransformDataType){
        dataTypeTransformConfigPath = YamlReader.getInstance(this.basicConfigPath).getString("dataTypeTransformConfigPath")
    }
    val needSaveToHive: Boolean = YamlReader.getInstance(this.basicConfigPath).getBoolean("needSaveToHive", false)
    def calc(): Unit={
        val calcOp = this.dataSource match {
            case "excel" => new CalculateV1(appName=this.appName,
                masterSet=this.masterSet, logLevel=this.logLevel,
                dataExportPath=this.dataExportPath, dataSavePath=this.dataSavePath,
                dataSaveStyle=this.dataSaveStyle, configPath=this.configPath, dataSource=this.dataSource)
            case "hive" => new CalculateV1(appName = this.appName,
                masterSet = this.masterSet, logLevel = this.logLevel,
                tableName=this.tableName, databaseName=this.databaseName,
                dataSavePath = this.dataSavePath, dataSaveStyle = this.dataSaveStyle,
                configPath = this.configPath, dataSource = this.dataSource)
            case _ => null //todo:补充mysql的
        }
//        calcOp.withSpark{session=>
//            {
//                if (!this.needTransformDataType){
//                    calcOp.metricsCalculateAndRepository(calcOp.getDataset())
//                }
//                else{
//                    calcOp.metricsCalculateAndRepository(this.transformDataType(calcOp.getDataset()))
//                }
//            }
//        }
//        if (!this.needTransformDataType) {
//            calcOp.metricsCalculateAndRepository(calcOp.getDataset())
//        }
//        else {
//            calcOp.metricsCalculateAndRepository(this.transformDataType(calcOp.getDataset()))
//        }
    }
//    def transformDataType(data: DataFrame): DataFrame={
//        val dataOverallOp = new DataOverallOp(configPath = this.dataTypeTransformConfigPath, this.needSaveToHive) //集群上需要改成true
//        dataOverallOp.columnTypeTransform(data)
//    }
}

object DeequCalculate extends App {
}
